package com.whoami.dataplatform.tag_combine;

import com.yahoo.memory.Memory;
import com.yahoo.sketches.kll.KllFloatsSketch;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileAsBinaryInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileAsBinaryOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Properties;

/**
 * 研究所有8千万种tidbits对应的uid数量的分布
 * DoublesSketch
 *
 * @author tzp
 * @since 2019/7/31
 */
@SuppressWarnings("Duplicates")
public class TidbitsUidQuantile extends Configured implements Tool {
    /**
     * 给MR框架的一堆配置, 文件里面的内容会被set到mr conf里
     */
    public static final String PARAMS_MR_PROP = "mr_properties";
    public static final int K = 2048;

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        String mrProFile = conf.get(PARAMS_MR_PROP);
        if (StringUtils.isNotEmpty(mrProFile)) {
            Properties mrProperties = new Properties();
            mrProperties.load(new FileInputStream(mrProFile));
            for (Object key : mrProperties.keySet()) {
                conf.set(key.toString(), mrProperties.get(key).toString());
            }
        }
        String date = args[0];//20190723
        String input = args[1];
        String output = args[2];//outputbase

        Job job = Job.getInstance(conf, this.getClass().getSimpleName() + " " + date);
        job.setJarByClass(this.getClass());

        //输入
        job.setInputFormatClass(SequenceFileAsBinaryInputFormat.class);
        FileInputFormat.addInputPath(job, new Path(input));

        //mapper
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(BytesWritable.class);

        //reducer
        job.setNumReduceTasks(1);
        job.setReducerClass(MyReducer.class);

        //输出
        job.setOutputFormatClass(SequenceFileAsBinaryOutputFormat.class);
        FileOutputFormat.setOutputPath(job, new Path(output));


        boolean result = job.waitForCompletion(true);

        return result ? 0 : 1;
    }

    public static class MyMapper extends Mapper<BytesWritable, BytesWritable, NullWritable, BytesWritable> {
        private KllFloatsSketch sketch;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            sketch = new KllFloatsSketch(K);
        }

        @Override
        protected void map(BytesWritable key, BytesWritable value, Context context) throws IOException, InterruptedException {
            ByteBuffer byteBuffer = ByteBuffer.wrap(value.getBytes());
            int num = byteBuffer.getInt();
            sketch.update(num);
            if (num > 1000) {
                context.getCounter("distribute", ">1000").increment(1L);
            }
            if (num > 2000) {
                context.getCounter("distribute", ">2000").increment(1L);
            }
            if (num > 10000) {
                context.getCounter("distribute", ">10000").increment(1L);
            }
            if (num > 200000) {
                context.getCounter("distribute", ">200000").increment(1L);
            }
            context.getCounter("distribute", "total").increment(1L);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            context.write(NullWritable.get(), new BytesWritable(sketch.toByteArray()));
        }
    }

    public static class MyReducer extends Reducer<NullWritable, BytesWritable, BytesWritable, BytesWritable> {
        KllFloatsSketch sketch = null;

        @Override
        protected void reduce(NullWritable key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {
            for (BytesWritable value : values) {
                KllFloatsSketch newSketch = KllFloatsSketch.heapify(Memory.wrap(value.copyBytes()));
                if (sketch == null) {
                    sketch = newSketch;
                } else {
                    sketch.merge(newSketch);
                }
            }
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            context.write(new BytesWritable(), new BytesWritable(sketch.toByteArray()));
        }
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Configuration(), new TidbitsUidQuantile(), args));
    }
}
